package com.ponhu.ea2.stockcenter.provider.config;

import com.alibaba.fastjson.JSON;
import com.ponhu.ea2.stockcenter.provider.constants.CommonConstant;
import com.ponhu.ea2.stockcenter.provider.entity.StockChangeMqLogs;
import com.ponhu.ea2.stockcenter.provider.service.entity.StockChangeMqLogsService;
import com.ponhu.ea2.stockcenter.provider.support.service.RabbitMQService;
import com.ponhu.ea2.stockcenter.vo.StockMqVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.BooleanUtils;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;

import java.util.List;
import java.util.stream.Collectors;

@Slf4j
@Configuration
public class RabbitMQConfig {

    //    @Value("${spring.rabbitmq.host}")
//    private String host;
//    @Value("${spring.rabbitmq.port}")
//    private int port;
    @Value("${spring.rabbitmq.username}")
    private String username;
    @Value("${spring.rabbitmq.password}")
    private String password;
    @Value("${spring.rabbitmq.addresses}")
    private String addresses;
    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;
    @Value("${spring.rabbitmq.publisher-confirms}")
    private Boolean publisherConfirms;
    @Value("${spring.rabbitmq.publisher-returns}")
    private Boolean publisherReturns;

    @Autowired
    private RabbitMQService rabbitMQService;
    @Autowired
    private StockChangeMqLogsService stockChangeMqLogsService;

    @Bean
    public CachingConnectionFactory cachingConnectionFactory() {
//        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(addresses);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        connectionFactory.setPublisherConfirms(publisherConfirms);
        connectionFactory.setPublisherReturns(publisherReturns);
        return connectionFactory;
    }

    @Bean
    @DependsOn("cachingConnectionFactory")
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory cachingConnectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setMandatory(true);
        // 和 publisherConfirms 不可共存
//        rabbitTemplate.setChannelTransacted(true);
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (BooleanUtils.isNotTrue(ack)) {
                ReturnedMessage returned = correlationData.getReturned();
                String jsonMsg = new String(returned.getMessage().getBody());
                log.info("消息发送未成功抵达服务器 ===>> ack={}, exchange={}, route={}, msg={}, cause={}",
                        ack, jsonMsg, returned.getExchange(), returned.getRoutingKey(), cause);

                sendMsgFail(returned, "消息发送未成功抵达服务器");
            }
        });
        rabbitTemplate.setReturnsCallback(returned -> {
            String jsonMsg = new String(returned.getMessage().getBody());
            log.info("消息发送未成功抵达消息队列 ===>> exchange={}, route={}, message={}",
                    returned.getExchange(), returned.getRoutingKey(), jsonMsg);

            sendMsgFail(returned, "消息发送未成功抵达消息队列");
        });
        return rabbitTemplate;
    }

    private void sendMsgFail(ReturnedMessage returned, String exceptionMsg) {

        String jsonMsg = new String(returned.getMessage().getBody());
        List<StockChangeMqLogs> mqLogsList = JSON.parseArray(jsonMsg, StockMqVO.class).stream()
                .map(mqVo -> new StockChangeMqLogs()
                        .setInventoryCode(mqVo.getInventoryCode())
                        .setExchange(returned.getExchange())
                        .setRoutingKey(returned.getRoutingKey()).setMessage(jsonMsg)
                        .setStatus(CommonConstant.SEND_MQ_STATUS_0))
                .collect(Collectors.toList());

        if (!stockChangeMqLogsService.saveBatch(mqLogsList))
            log.error("货品库存销售库存变动发送消息保存日志异常 ===>> {}", JSON.toJSONString(mqLogsList));

        rabbitMQService.mqFailSendDingTalk(JSON.toJSONString(mqLogsList), exceptionMsg);
    }

    /**
     * 针对消费者配置
     * 1. 设置交换机类型
     * 2. 将队列绑定到交换机
     * FanoutExchange：将消息分发到所有的绑定队列，无routingkey的概念
     * HeadersExchange：通过添加属性key-value匹配
     * DirectExchange：按照routingkey分发到指定队列
     * TopicExchange：多关键字匹配
     *
     * @return
     */
    @Bean
    public TopicExchange saleStockTopicExchange() {
        return new TopicExchange(CommonConstant.SALE_STOCK_TOPIC_EXCHANGE);
    }

    @Bean
    public Queue xcxQueue() {
        return new Queue(CommonConstant.XCX_QUEUE, true); //队列持久
    }

//    @Bean
    public Queue tbQueue() {
        return new Queue(CommonConstant.TB_QUEUE, true); //队列持久
    }

    @Bean
    public Binding bindingSaleStockTopicExchange2XcxQueue(Queue xcxQueue, TopicExchange saleStockTopicExchange) {
        return BindingBuilder.bind(xcxQueue).to(saleStockTopicExchange).with(CommonConstant.XCX_ROUTINGKEY);
    }

//    @Bean
    public Binding bindingSaleStockTopicExchange2TbQueue(Queue tbQueue, TopicExchange saleStockTopicExchange) {
        return BindingBuilder.bind(tbQueue).to(saleStockTopicExchange).with(CommonConstant.TB_ROUTINGKEY);
    }

}
